package cn.doitedu.api;


import beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.time.Duration;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/25
 * @Desc: 学大数据，上多易教育
 * 从kafka接收用户行为数据
 * {"session_id":"s01","uid":1,"timestamp":1708743988000,"event_id":"A"}
 * <p>
 * 实时监控每个用户：
 * 如果一个用户发生  K事件 >=3 次,   [Y R Q ]序列 >= 2次 ，
 * 前面两个条件满足之后，当再发生 X行为，则延迟 30s 发出一条消息 {"user_id":1,"timestamp":1708743988000,"message":"ok"}
 * 而且，对一个用户，在 2小时内 最多发送 3条消息
 **/
public class _19_Exercise_ZongHe {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setStateBackend(new HashMapStateBackend());
        //env.setStateBackend(new EmbeddedRocksDBStateBackend(true));


        // 构建source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setGroupId("gtt")
                .setClientIdPrefix("doit46-")
                .setBootstrapServers("doitedu:9092")
                .setTopics("doit46-2")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream =
                env.fromSource(source,
                        WatermarkStrategy
                                .<String>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                                    @Override
                                    public long extractTimestamp(String json, long recordTimestamp) {
                                        JSONObject jsonObject = JSON.parseObject(json);
                                        return jsonObject.getLongValue("timestamp");
                                    }
                                })
                                .withIdleness(Duration.ofMillis(500)),   // 当多个数据来源中某个来源idle超过指定时间，生成合并watermark时就会忽略这个来源
                        "kfk", TypeInformation.of(String.class));


        // 先把json数据解析成javabean
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));


        // 必须让相同用户的行为数据发送到下游相同的运算并行度（subTask）
        KeyedStream<UserAction, Integer> keyedStream = mapped.keyBy(UserAction::getUid);


        // 然后开始组织核心逻辑
        keyedStream.process(new KeyedProcessFunction<Integer, UserAction, String>() {

            ValueState<Integer> cntState;
            ValueState<Integer> idxState;
            ValueState<Integer> seqCntState;
            String[] eventSeq = {"Y", "R", "Q"};

            ListState<Long> msgHistoryState;

            @Override
            public void open(Configuration parameters) throws Exception {

                cntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("cnt", Integer.class));
                idxState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("idx_cnt", Integer.class));
                seqCntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("seq_cnt", Integer.class));


                ListStateDescriptor<Long> desc = new ListStateDescriptor<>("msg-history", Long.class);
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.hours(2)).neverReturnExpired().build();

                desc.enableTimeToLive(ttlConfig);
                msgHistoryState = getRuntimeContext().getListState(desc);

            }

            @Override
            public void processElement(UserAction userAction, KeyedProcessFunction<Integer, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                // K事件 >=3 次,
                if ("K".equals(userAction.getEvent_id())) {
                    int oldCnt = cntState.value() == null ? 0 : cntState.value();
                    cntState.update(oldCnt + 1);

                    System.out.println("当前K次数： " + cntState.value());
                }

                // [Y R Q ]序列 >= 2次
                // 取出当前正在匹配的事件是哪一个
                int idx = idxState.value() == null ? 0 : idxState.value();
                if (userAction.getEvent_id().equals(eventSeq[idx])) {
                    // 如果匹配上cntState。
                    int nextIdx = idx + 1;
                    // 如果idx再加1已经超出事件序列长度，则回归到0，并更新序列完成次数
                    if (nextIdx > 2) {
                        nextIdx = 0;
                        seqCntState.update(seqCntState.value() == null ? 1 : seqCntState.value() + 1);
                        System.out.println("nextIdx超过2了，重置为：0，seq完成当前次数: " + seqCntState.value());

                    }
                    idxState.update(nextIdx);
                    System.out.println("nextIdx： " + idxState.value());
                }


                // 判断输入的事件，是否是X
                if (userAction.getEvent_id().equals("X")) {
                    // 判断该用户上述两个统计条件是否已经满足
                    // 先取出K事件统计到的次数，YRQ序列统计到的次数
                    int kCount = cntState.value() == null ? 0 : cntState.value();
                    int seqCount = seqCntState.value() == null ? 0 : seqCntState.value();

                    System.out.println("X事件到达，取到的k次数和seq次数分别是：" + kCount + "," + seqCount);

                    if (kCount >= 3 && seqCount >= 2) {

                        // 判断现在是否该发消息
                        long historyStart = userAction.getTimestamp() - 2*60*60*1000L;
                        long historyEnd = userAction.getTimestamp();

                        // 去消息发送历史记录中查询这段时间范围内一共有几条
                        int i = 0;
                        for (Long msgTimeStamp : msgHistoryState.get()) {
                            if(msgTimeStamp>= historyStart && msgTimeStamp<= historyEnd){
                                i++;
                            }
                        }
                        System.out.println("历史记录中找到了消息发送条数为: " + i);

                        if(i<3) {
                            // 准备发消息(注册一个30s后的定时器)
                            ctx.timerService().registerEventTimeTimer(userAction.getTimestamp() + 30000);
                            // 将发送消息的时间记录到发送历史状态中
                            msgHistoryState.add(userAction.getTimestamp());
                        }
                    }
                }
            }


            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<Integer, UserAction, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                JSONObject messageObject = new JSONObject();
                messageObject.put("user_id", ctx.getCurrentKey());
                messageObject.put("time_stamp", timestamp);
                messageObject.put("msg", "ok");

                out.collect(messageObject.toJSONString());
            }
        }).print();

        env.execute();


    }
}
